package dag

import (
	"errors"

	"github.com/shiningrush/fastflow/pkg/entity"
	"github.com/shiningrush/fastflow/pkg/mod"
	"github.com/shiningrush/fastflow/pkg/utils/data"
	"gopkg.in/yaml.v3"
)

func NewDagGraph(yamlfile string) error {
	dag := entity.Dag{
		Status: entity.DagStatusNormal,
	}
	err := yaml.Unmarshal([]byte(yamlfile), &dag)
	if err != nil {
		return err
	}

	oDag, err := mod.GetStore().GetDag(dag.ID)
	if err != nil && !errors.Is(err, data.ErrDataNotFound) {
		return err
	}
	if oDag != nil {
		return mod.GetStore().UpdateDag(&dag)
	}

	return mod.GetStore().CreateDag(&dag)
}

func RunDAGTask(dagname string) error {
	_, err := mod.GetCommander().RunDag(dagname, nil)
	return err
}
